<?php

namespace Rabbitmq\Tool\RabbitMQTool;

use InvalidArgumentException;
use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Connection\AbstractConnection;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

class RabbitMQInstance
{
    /**
     * 队列
     * @var string
     */
    protected $queue = '';

    /**
     * 交换机
     * @var string
     */
    protected $exchange = '';

    /**
     * AMQPExchange Type
     * 交换机类型
     *
     * @var string
     */
    protected $exchangeType = '';

    /**
     * Consumer identifier
     * 用户标签
     *
     * @var string
     */
    protected $consumerTag = '';

    /**
     * AMQPMessage Route Key
     * 路由键
     *
     * @var string
     */
    protected $routeKey = '';

    /**
     * AMQPMessage vhost
     *
     * @var string|null
     */
    protected $vhost = null;

    /**
     * @var bool
     * true
     *  1.如果queue已存在 则直接连接；
     *  2.如果queue不存在 则直接报错
     * false
     *  1.如果queue不存在 则创建新的queue
     *  2.如果queue已存在 会报错。
     */
    protected $passive = false;

    /**
     * @var bool
     * true
     *  1.如果exchange已存在 则直接连接 并且不检查配置 比如已存在的exchange是fanout，新需要建立的是direct，也不会报错；
     *  2.如果exchange不存在 则直接报错
     * false
     *  1.如果exchange不存在 则创建新的exchange
     *  2.如果exchange已存在 则判断配置是否相同。如果配置不相同 则直接报错。比如已存在的exchange是fanout，新需要建立的是direct，会报错。
     */
    protected $exchangePassive;

    /**
     * 设置是否持久化。设置true表示持久化，反之非持久化。持久化可以将交换器存盘，在服务器重启的时候不会丢失相关信息。
     * @var bool
     */
    protected $durable = false;

    /**
     * 设置是否排他。设置true时，表示为排他队列，该队列仅对首次首次声明它的连接可见，并在连接断开时自动删除。
     * 注意三点：排他队列是基于连接（Connection）可见的，
     * 同一个连接的不同信道（Channel）是可以同时访问同一连接创建的排他队列；
     * “首次”是指如果同一个连接已经声明了一个排他队列，其他连接时不允许建立同名排他队列的，
     * 这个与普通队列不同；即使该队列是持久化的，一旦连接关闭或客户端退出，该排他队列都会自动删除，这种队列适合一个客户端同时发送和读取消息的场景
     * @var bool
     */
    protected $exclusive = false;

    /**
     * @var
     * true 当最后一个消费者取消订阅之后 exchange会被自动删除 一般用于临时exchange
     * false 不自动删除交换机
     */
    protected $autoDelete = false;

    /**
     * Auto Ack
     * 自动消费
     *
     * @var bool
     */
    protected $autoAck = false;

    /**
     * 发送消息负载
     * @var string
     */
    protected $payload = '';

    /**
     * 消息类型属性
     * @var
     */
    private $properties;

    /**
     * 异步执行，为true时，不等待队列创建结果，立即完成函数调用
     * @var
     */
    private $noWait = false;

    /**
     * 设定消息队列的额外参数，如存活时间等
     * @var array
     *  'x-max-length'
     消息条数限制,该参数是非负整数值。限制加入queue中消息的条数。先进先出原则，超过10条后面的消息会顶替前面的消息。
     *  'x-max-length-bytes'
     消息容量限制,该参数是非负整数值。该参数和x-max-length目的一样限制队列的容量，但是这个是靠队列大小（bytes）来达到限制。
     *  'x-message-ttl'
     消息存活时间,该参数是非负整数值.创建queue时设置该参数可指定消息在该queue中待多久，可根据x-dead-letter-routing-key和
     x-dead-letter-exchange生成可延迟的死信队列。
     *  'x-max-priority'
     消息优先级,创建queue时arguments可以使用x-max-priority参数声明优先级队列 。该参数应该是一个整数，表示队列应该支持的最大优先级。
     建议使用1到10之间。目前使用更多的优先级将消耗更多的资源（Erlang进程）设置该参数同时设置死信队列时或造成已过期的低优先级消息会在未过期的
     高优先级消息后面执行。该参数会造成额外的CPU消耗。
     *  'x-expires'
     存活时间,创建queue时参数arguments设置了x-expires参数，该queue会在x-expires到期后queue消息，亲身测试直接消失（哪怕里面有未消费的消息）
     *  'x-dead-letter-exchange和x-dead-letter-routing-key'
     创建queue时参数arguments设置了x-dead-letter-routing-key和x-dead-letter-exchange，会在x-message-ttl时间到期后把消息放
     到x-dead-letter-routing-key和x-dead-letter-exchange指定的队列中达到延迟队列的目的。
     */
    private $arguments = array();

    /**
     * 创建链接
     */
    private function getConnection(): AMQPStreamConnection
    {
        $conf = config('rabbitmq.config');
        //建立生产者与mq之间的连接
        return new AMQPStreamConnection($conf['host'], $conf['port'], $conf['user'], $conf['password'], $conf['vhost']);
    }

    /**
     * 交换机绑定
     * @param $exchange
     * @return $this
     */
    public function setExchange($exchange): RabbitMQInstance
    {
        $this->exchange = $exchange;
        return $this;
    }

    /**
     * 交换机绑定
     * @param $routeKey
     * @return $this
     */
    public function setRouteKey($routeKey): RabbitMQInstance
    {
        $this->routeKey = $routeKey;
        return $this;
    }

    /**
     * 交换机类型设置
     * @param $exchangeType
     * @return $this
     */
    public function setExchangeType($exchangeType): RabbitMQInstance
    {
        $this->exchangeType = $exchangeType;
        return $this;
    }

    /**
     * 被动模式，为true时，如果$queue不存在，则返回错误(不创建队列，只是检测队列是否存在)，为false时，如果$queue不存在，则创建此队列，然后返回OK
     * @param $passive
     * @return $this
     */
    public function setQueuePassive($passive): RabbitMQInstance
    {
        $this->passive = $passive;
        return $this;
    }

    /**
     * @param $passive
     * @return $this
     */
    public function setExchangePassive($passive): RabbitMQInstance
    {
        $this->exchangePassive = $passive;
        return $this;
    }

    /**
     * @param $autoAck
     * @return $this
     */
    public function setAutoAck($autoAck): RabbitMQInstance
    {
        $this->autoAck = $autoAck;
        return $this;
    }

    /**
     * 排他性，为true时，只能在本次连接中使用，连接关闭时自动消亡(即使$durable为true也是)
     * @param $exclusive
     * @return $this
     */
    public function setExclusive($exclusive): RabbitMQInstance
    {
        $this->exclusive = $exclusive;
        return $this;
    }

    /**
     * @param $noWait
     * @return $this
     */
    public function setNoWait($noWait): RabbitMQInstance
    {
        $this->noWait = $noWait;
        return $this;
    }

    /**
     * @param $arguments
     * @return $this
     */
    public function setArguments($arguments): RabbitMQInstance
    {
        $this->arguments = $arguments;
        return $this;
    }

    /**
     * 持久化设置: 为true时，消息将存入数据库，即使服务崩溃，消息也不会消失
     * @param $durable
     * @return $this
     */
    public function setDurable($durable): RabbitMQInstance
    {
        $this->durable = $durable;
        return $this;
    }

    /**
     * 自动删除设置
     * @param $autoDelete
     * @return $this
     */
    public function setAutoDelete($autoDelete): RabbitMQInstance
    {
        $this->autoDelete = $autoDelete;
        return $this;
    }

    /**
     * 队列设置
     * @param $queue
     * @return $this
     */
    public function setQueue($queue): RabbitMQInstance
    {
        $this->queue = $queue;
        return $this;
    }

    /**
     * 队列负载
     * @param $payload
     * @return $this
     */
    public function setPayload($payload): RabbitMQInstance
    {
        $this->payload = $payload;
        return $this;
    }

    /**
     * 队列属性
     * @param $properties
     * @return $this
     */
    public function setProperties($properties): RabbitMQInstance
    {
        $this->properties = $properties;
        return $this;
    }

    /**
     * get queue name
     *
     * @return string
     */
    public function getQueue(): string
    {
        if (empty($this->queue)) {
            throw new InvalidArgumentException("'queue' key is required.");
        }
        return $this->queue;
    }

    /**
     * Get exchange name
     *
     * @return string
     */
    public function getExchange(): string
    {
        if (empty($this->exchange)) {
            throw new InvalidArgumentException("'exchange' key is required.");
        }
        return $this->exchange;
    }

    /**
     * Get exchange name
     *
     * @return string
     */
    public function getExchangeType(): string
    {
        if (empty($this->exchangeType)) {
            throw new InvalidArgumentException("'exchange type' key is required.");
        }
        return $this->exchangeType;
    }

    /**
     * Get Route Key
     *
     * @return string
     */
    public function getRouteKey(): string
    {
        return $this->routeKey;
    }

    /**
     * $exchange
     * $payload
     * $properties
     * @throws \Exception
     */
    public function publish()
    {
        list($connection, $channel) = $this->startConnection();
        $message = new AMQPMessage($this->payload, $this->properties);
        $channel->basic_publish($message, $this->exchange);
        $this->close($connection, $channel);
    }

    /**
     * Amqp Message Consumer
     *
     * @param callable $processMessage
     * @throws \Exception
     */
    public function consume(callable $processMessage)
    {
        list($connection, $channel) = $this->startConnection();
        $channel->basic_qos(null, 3, null);
        $channel->basic_consume($this->queue, $this->consumerTag, false, false, false, false,
            function (AMQPMessage $message) use ($processMessage) {
                return $processMessage($message);
        });
        while ($channel->is_consuming()) {
            $channel->wait();
        }
        $this->close($connection, $channel);
    }

    /**
     * Start Connection
     *
     * @return array
     * @throws \Exception
     */
    protected function startConnection(): array
    {
        $connection = $this->getConnection();
        $channel = $connection->channel();
        /**
         * name: $exchange
         * type: direct
         * passive: false
         * durable: true the exchange will survive server restarts
         * auto_delete: false the exchange won't be deleted once the channel is closed.
         */
        $channel->exchange_declare($this->getExchange(), $this->getExchangeType(), $this->exchangePassive, $this->durable, $this->autoDelete);
        /**
         * The following code is the same both in the consumer and the producer.
         * In this way we are sure we always have a queue to consume from and an
         * exchange where to publish messages.
         *
         * name: $queue
         * passive: false
         * durable: true the queue will survive server restarts
         * exclusive: false the queue can be accessed in other channels
         * auto_delete: false the queue won't be deleted once the channel is closed.
         */
        $channel->queue_declare($this->getQueue(), $this->passive, $this->durable, $this->exclusive, $this->autoDelete, $this->noWait, $this->arguments);
        $channel->queue_bind($this->getQueue(), $this->getExchange(), $this->getRouteKey());
        return [$connection, $channel];
    }

    /**
     * @param $connection
     * @param $channel
     */
    public function close($connection, $channel)
    {
        $connection->close();
        $channel->close();
    }
}
